
#ifndef __IPC_SESSION_HPP__
#define __IPC_SESSION_HPP__

#include "khan.hpp"
#include "io_session.hpp"
#include "ipc_message.hpp"

typedef std::deque<ipc_message> ipc_message_queue;

typedef boost::function<void(const ipc_message msg)> WriteFunction;

template <typename WorkerPtr, typename Manager>
class ipc_session
: public io_session,
public boost::enable_shared_from_this<ipc_session<WorkerPtr, Manager> >,
private boost::noncopyable       
{
public:
    ipc_session(int id, boost::asio::io_service& io_service, 
            Manager& manager,  WorkerPtr& work)
        : _id(id), 
        _socket(io_service), _manager(manager), 
        _worker(work)
    { 
        PDBG("start session id: %d\n", _id);
    }
    ~ipc_session() { printf("bye bye ipc_session\n"); }

    tcp_socket& socket() { return _socket; }

    int id() { return _id; }
    void start()
    {
        _worker->reg_write_func(
                boost::bind(&ipc_session::send_ipc, 
                    this->shared_from_this(), _1));
        boost::asio::async_read(_socket,
                boost::asio::buffer(_read_mesg.data(), ipc_message::header_length),
                boost::bind(
                    &ipc_session::handle_read_header, this->shared_from_this(),
                    boost::asio::placeholders::error));
    }

    void stop()     /* FIXME... plz... */
    { 
        PDBG("stop this session\n");
        boost::system::error_code error;
        _socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, error);   /* XXX: */
        _worker.reset();    /*  FIXME */
        _manager.leave(this->shared_from_this());
    }    

    void send_ipc(const ipc_message& msg)
    {
        bool write_in_progress = !_write_msgs.empty();
        _write_msgs.push_back(msg);
        if (!write_in_progress) {
            boost::asio::async_write(_socket,
                    boost::asio::buffer(_write_msgs.front().data(),
                        _write_msgs.front().length()),
                    boost::bind(&ipc_session::handle_write, this->shared_from_this(),
                        boost::asio::placeholders::error));
        }
    }

private:

    void handle_read_header(const boost::system::error_code& error)
    {
        if (!error && _read_mesg.decode_header())
        {
            boost::asio::async_read(_socket,
                    boost::asio::buffer(_read_mesg.body(), _read_mesg.body_length()),
                    boost::bind(&ipc_session::handle_read_body, this->shared_from_this(),
                        boost::asio::placeholders::error, 
                        boost::asio::placeholders::bytes_transferred
                        ));
        }
        else
        {
            PDBG("exception: %s\n", error.message().c_str());
            this->stop(); 
        }
    }

    void handle_read_body(const boost::system::error_code& error, 
            std::size_t bytes_transferred)
    {
        if (!error)
        {
            PDBG("received body size: %d", (int)bytes_transferred);
            _worker->handle_ingress_msg(_read_mesg);
            boost::asio::async_read(_socket,
                    boost::asio::buffer(_read_mesg.data(), ipc_message::header_length),
                    boost::bind(&ipc_session::handle_read_header, this->shared_from_this(),
                        boost::asio::placeholders::error));
        }
        else
        {
            PDBG("exception: %s\n", error.message().c_str());
            this->stop();
        }
    }

    void handle_write(const boost::system::error_code& error)
    {
        if (!error)
        {
            _write_msgs.pop_front();
            if (!_write_msgs.empty())
            {
                boost::asio::async_write(_socket,
                        boost::asio::buffer(_write_msgs.front().data(),
                            _write_msgs.front().length()),
                        boost::bind(&ipc_session::handle_write, this->shared_from_this(),
                            boost::asio::placeholders::error));
            }
        }
        else
        {
            PDBG("exception: %s\n", error.message().c_str());
            this->stop(); 
        }
    }

    int _id;
    tcp_socket   _socket;
    Manager&     _manager;
    WorkerPtr    _worker;
    ipc_message _read_mesg;
    ipc_message_queue _write_msgs;
};

#endif
